#include "ThreadPool.h"
#include "NamedIndex.h"
#include "SocketServer.h"


#include "Utils.h"
#include "Exception.h"
#include "NamedIndex.h"
using JetByteTools::Win32::CThreadPool;
using JetByteTools::Win32::CNamedIndex;
#include "TangoThreadPool.h"
#include "TangoThreadPoolWorkerThread.h"
using JetByteTools::Win32::CIOCompletionPort;
using JetByteTools::Win32::CEvent;
using JetByteTools::Win32::Output;
//using JetByteTools::Win32::CSocketServer;
using JetByteTools::Win32::CIOBuffer;
using JetByteTools::Win32::CException;
using JetByteTools::Win32::CNamedIndex;
using JetByteTools::Win32::_tstring;
using JetByteTools::Win32::ToString;

typedef JetByteTools::Win32::CSocketServer::Socket Socket;
CTangoThreadPool::CTangoThreadPool(
	CNamedIndex &userDataSlots,
	size_t initialThreads,
	size_t minThreads,
	size_t maxThreads,
	size_t maxDormantThreads,
	size_t poolMaintPeriod,
	size_t dispatchTimeout,
	size_t maxThreadsDispatchTimeout,
	size_t numReads,
	bool usingSequenceNumbers,
	bool useMultipleWrites)
	: JetByteTools::Win32::CThreadPool(
	initialThreads,
	minThreads,
	maxThreads,
	maxDormantThreads,
	poolMaintPeriod,
	dispatchTimeout,
	maxThreadsDispatchTimeout),
	m_numReads(numReads),
	m_usingSequenceNumbers(usingSequenceNumbers),
	m_useMultipleWrites(useMultipleWrites),
	m_userDataIndex(userDataSlots.Add(_T("CThreadPool::PerConnectionDataIndex")))
{
}
JetByteTools::Win32::CThreadPool::WorkerThread *CTangoThreadPool::CreateWorkerThread(
	CIOCompletionPort &iocp,
	CEvent &messageReceivedEvent,
	JetByteTools::Win32::CThreadPool &pool)
{
	return new CTangoThreadPoolWorkerThread(
		iocp,
		messageReceivedEvent,
		*this);
}

void CTangoThreadPool::OnThreadStarted()
{
	//global::LogMess(_T("OnThreadStarted"));
}

void CTangoThreadPool::OnThreadInitialised()
{
	//global::LogMess(_T("OnThreadInitialised"));
}

void CTangoThreadPool::OnProcessingStarted()
{
	//global::LogMess(_T("OnProcessingStarted"));
}

void CTangoThreadPool::OnReceivedMessage()
{
	//global::LogMess(_T("OnReceivedMessage"));
}

void CTangoThreadPool::OnProcessingStopped()
{
	//global::LogMess(_T("OnProcessingStopped"));
}

void CTangoThreadPool::OnThreadStopped()
{
	//global::LogMess(_T("OnThreadStopped"));
}

void CTangoThreadPool::DispatchConnectionEstablished(
	Socket *pSocket,
	CIOBuffer *pAddress)
{
	// Allocate per connection data

	CPerConnectionData *pData = new CPerConnectionData();

	// And store our data in the socket...

	pSocket->SetUserPtr(m_userDataIndex, pData);

	DoDispatch(pSocket, pAddress, ConnectionEstablished);
}

void CTangoThreadPool::DispatchConnectionClosing(
	Socket *pSocket)
{
	DoDispatch(pSocket, 0, ConnectionClosing);
}

void CTangoThreadPool::DispatchReadCompleted(
	Socket *pSocket,
	CIOBuffer *pBuffer)
{
	DoDispatch(pSocket, pBuffer, ReadCompleted);
}

void CTangoThreadPool::DoDispatch(
	Socket *pSocket,
	CIOBuffer *pBuffer,
	DispatchEvents event)
{
	pSocket->AddRef();

	if (pBuffer)
	{
		pBuffer->AddRef();
	}
	else if (event != ConnectionClosing)
	{
		throw CException(_T("CThreadPool::DoDispatch()"), _T("Unexpected: Only the ConnectionClosing event can have a null pBuffer"));
	}

	Dispatch(
		reinterpret_cast<ULONG_PTR>(pSocket),
		event,
		pBuffer);
}

void CTangoThreadPool::OnSocketReleased(
	Socket *pSocket)
{
	// release per connection data

	CPerConnectionData *pData = GetPerConnectionData(pSocket);

	pSocket->SetUserPtr(m_userDataIndex, 0);

	delete pData;
}


CTangoThreadPool::CPerConnectionData *CTangoThreadPool::GetPerConnectionData(
	Socket *pSocket) const
{
	return static_cast<CPerConnectionData*>(pSocket->GetUserPtr(m_userDataIndex));
}

///////////////////////////////////////////////////////////////////////////////
// CThreadPool::CPerConnectionData
///////////////////////////////////////////////////////////////////////////////

CTangoThreadPool::CPerConnectionData::CPerConnectionData()
	: m_outOfSequenceReads(m_crit),
	m_numUses(0)
{
}

void CTangoThreadPool::CPerConnectionData::SetAddress(
	const _tstring &address)
{
	m_address = address;
}

void CTangoThreadPool::CPerConnectionData::UseConnection()
{
	//lint -e{534} Ignoring return value of function 
	::InterlockedIncrement(&m_numUses);
}

_tstring CTangoThreadPool::CPerConnectionData::GetConnectionDetails() const
{
	return m_address + _T(" - ") + ToString(m_numUses);
}